package sunyu.tools.activemq;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.pool.PooledConnection;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.jms.JMSException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ActiveMQ工具类
 * <p>
 * 维护了多个链接、生产者和消费者
 *
 * @author 孙宇
 */
@Component
public class ActiveMQTools {

    public static final ConcurrentHashMap<String, PooledConnection> pooledConnectionMap = new ConcurrentHashMap<>();
    public static final ConcurrentHashMap<String, ActiveMQSenderTemplate> senderMap = new ConcurrentHashMap<>();
    public static final ConcurrentHashMap<String, ActiveMQReceiverTemplate> receiverMap = new ConcurrentHashMap<>();
    public static final String defaultName = "main";
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @PreDestroy
    public void destroy() {
        logger.info("销毁ActiveMQ所有连接池开始");
        for (Map.Entry<String, PooledConnection> entry : pooledConnectionMap.entrySet()) {
            try {
                entry.getValue().close();
            } catch (JMSException e) {
                logger.error(ExceptionUtils.getStackTrace(e));
            }
        }
        logger.info("销毁ActiveMQ所有连接池结束");
    }

    public void addSender(ActiveMQSenderTemplate sender) {
        logger.info("添加Sender：{}", sender.getName());
        if (senderMap.containsKey(sender.getName())) {
            logger.error("{}已存在!", sender.getName());
        } else {
            senderMap.put(sender.getName(), sender);
        }
    }

    public ActiveMQSenderTemplate getSender(String name) {
        return senderMap.get(name);
    }

    public void addReceiver(ActiveMQReceiverTemplate receiver) {
        logger.info("添加Receiver：{}", receiver.getName());
        if (receiverMap.containsKey(receiver.getName())) {
            logger.error("{}已存在!", receiver.getName());
        } else {
            receiverMap.put(receiver.getName(), receiver);
        }
    }

    public ActiveMQReceiverTemplate getReceiver(String name) {
        return receiverMap.get(name);
    }

    public void addConnection(String connectionName,
                              String brokerUrl) {
        logger.info("创建连接池：{}", connectionName);
        if (pooledConnectionMap.containsKey(connectionName)) {
            logger.error("{}连接池已存在!", connectionName);
        } else {
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
            activeMQConnectionFactory.setUserName(ActiveMQConnection.DEFAULT_USER);
            activeMQConnectionFactory.setPassword(ActiveMQConnection.DEFAULT_PASSWORD);
            activeMQConnectionFactory.setBrokerURL(brokerUrl);
            activeMQConnectionFactory.setDispatchAsync(true);//异步发送消息
            PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(activeMQConnectionFactory);
            pooledConnectionFactory.setMaximumActiveSessionPerConnection(200);
            pooledConnectionFactory.setIdleTimeout(120);
            pooledConnectionFactory.setMaxConnections(5);
            pooledConnectionFactory.setBlockIfSessionPoolIsFull(true);
            try {
                PooledConnection connection = (PooledConnection) pooledConnectionFactory.createConnection();
                connection.start();
                pooledConnectionMap.put(connectionName, connection);
            } catch (JMSException e) {
                logger.error(ExceptionUtils.getStackTrace(e));
            }
        }
    }

    public PooledConnection getConnection(String connectionName) {
        return pooledConnectionMap.get(connectionName);
    }

}
